package org.example.flink.windows;


import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Hello09WindowFunctionByReduce {
    public static void main(String[] args) throws Exception {
        //运行环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        //获取数据源-admin:3
        DataStreamSource<String> source = environment.socketTextStream("localhost", 19523);
        //CountWindow--Tumbling--增量计算
        source.map(word -> Tuple2.of(word.split(":")[0],
                        Integer.parseInt(word.split(":")[1])), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(tuple2 -> tuple2.f0)
                .countWindow(3)
                .reduce((t1, t2) -> {
                    System.out.println("窗口增量计算函数-来一条算一条.main[" + t1 + "][" + t2 + "]");
                    t1.f0 = t1.f0 + "_" + t2.f0;
                    t1.f1 = t1.f1 + t2.f1;
                    return t1;
                }).print("CountWindow--Tumbling:").setParallelism(1);
        //运行环境
        environment.execute();
    }
}
